﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Cinchoo.Core;
using System.ComponentModel;
using Cinchoo.Core.Net.Sockets;
using System.Threading.Tasks;
using System.Threading;
using ChoWorkflowServer.Model;
using System.Data.SqlClient;
using System.Net.Sockets;
using System.Net;
using Cinchoo.Core.Diagnostics;
using Cinchoo.Core.ETL;
using System.Reflection;
using System.Diagnostics;
using Cinchoo.Core.IO;
using Cinchoo.Core.Net;
using Cinchoo.Core.Shell;
using System.ServiceModel;
using System.ServiceModel.Description;
using System.IO;
using Cinchoo.Core.Threading.Tasks;
using Cinchoo.Core.Reflection;
using Cinchoo.Core.Compiler;
using Cinchoo.Core.Services;
using Cinchoo.IO;
using Cinchoo.Core.WPF;
using System.Windows;

namespace ChoWorkflowServer
{
    [ChoApplicationHost]
    public partial class ChoWorkflowServerEx : ChoApplicationHost
    {
        #region Instance Data Members (Private)

        public new static ChoWorkflowServerEx Instance;
        internal volatile bool ResetRequested = false;

        private ChoPingAppServer _pingAppServer = null;
        private ChoTcpServer _tcpServer = null;
        private ChoTcpClient _tcpClient = null;
        private ChoAbortableAsyncResult _msgBroadcastAsyncResult = null;
        private DateTime _modifiedDate = ChoDbDateTime.MinValue;
        private string _machineName = Environment.MachineName;
        private string _worklowServiceProgramFilePath;
        private ChoMemoryMappedFileReader<ChoCommandMessage> _mmr;
        private ChoETLWorkflow _runningWF;
        private DateTime _lastModifiedDate = DateTime.MinValue;
        private ChoWSControllerConnectionManager _wsControllerConnectionManager;
        private ServiceHost _wcfHost = null;
        private ChoIdleTask _idleTask = null;

        private readonly CancellationTokenSource _tokenSource = new CancellationTokenSource();
        private readonly CancellationTokenSource _cts = new CancellationTokenSource();
        private ChoWorkflowScheduler _workflowScheduler;
        private SemaphoreSlim _concurrencySemaphore;

        private readonly object _taskActionCommandLock = new object();
        private readonly object _mmwLock = new object();
        private readonly Dictionary<string, ChoMemoryMappedFileWriter<ChoCommandMessage>> _mmwDict = new Dictionary<string, ChoMemoryMappedFileWriter<ChoCommandMessage>>();

        private int _exitCode;

        private ChoAbortableAsyncResult _clientControllerAsyncResult = null;
        private IChoDataModel _dataModel;
        private ChoWorkflowServerSettings _workflowServerSettings;
        private ChoWFExecutionLineArgsObject _wfExecutionLineArgsObject = null;
        private ChoWFServerCmdLineArgsObject _wfServerCmdLineArgsObject = null;
        
        public bool MaxNoOfAllowedTasksReached { get; private set; }

        #endregion Instance Data Members (Private)

        public ChoWorkflowServerEx()
        {
            Instance = this;
        }

        #region Main Entry

        public void Run(string[] args)
        {
            try
            {
                //ChoApplication.PropertyResolve += new EventHandler<ChoUnknownProperyEventArgs>(ChoApplication_PropertyResolve);

                ChoUser.CurrentUser = ChoUser.SYSTEM_ACCT;
                if (ChoWFServerCmdLineArgBuilder.RunAsServer)
                    ChoApplication.Run(args);
                else
                {
#if DEBUG
                    //Debugger.Launch();
#endif

                    ChoWFServerCmdLineArgsObject cmd = new ChoWFServerCmdLineArgsObject();
                    if (cmd.DebugMode)
                    {
                        cmd.RunAsPeer = true;
                        _dataModel = ChoDataModelFactory.New();
                    }
                    else
                    {
                        ChoWSUDPConnectionManager.Instance.StartAsPeer();
                        _workflowServerSettings = new ChoWorkflowServerSettings();
                        _dataModel = ChoActivator.CreateInstance(typeof(ChoWorkflowServer.ModelWCF.ChoDataModel)) as IChoDataModel;
                    }

                    ChoTrace.WriteLine("CommandLineArgs: {0}".FormatString(String.Join(" ", Environment.GetCommandLineArgs())));

                    RunWorkflow();

                    ChoFramework.Shutdown();
                    Environment.Exit(_exitCode);
                }
            }
            catch (ChoCommandLineArgUsageException cEx)
            {
                Console.WriteLine(cEx.Message);
                ChoTrace.Error(cEx);
                ChoFramework.Shutdown();
                Environment.Exit(-1);
            }
            catch (Exception ex)
            {
                if (_wfExecutionLineArgsObject != null && _wfExecutionLineArgsObject.BatchId != 0)
                {
                    SetTaskStatus(_wfExecutionLineArgsObject.BatchId, _wfExecutionLineArgsObject.TaskName, _wfExecutionLineArgsObject.ScheduleName, 
                        ChoETLWorkflowStatus.FAILED, _wfExecutionLineArgsObject.UserName, Environment.MachineName, ex.Message);
                }
                if (ChoWFServerCmdLineArgBuilder.RunAsServer/* || _wfServerCmdLineArgsObject.DebugMode*/)
                    Console.WriteLine(ex.ToString());
                ChoTrace.Error(ex);
                //ChoFramework.Shutdown();
                Environment.Exit(-1);
            }
        }

        protected override void ApplyGlobalApplicationSettingsOverrides(ChoGlobalApplicationSettings obj)
        {
            obj.LogSettings.DoAppendProcessIdToLogDir = false;
            obj.LogSettings.ExcludeLogMsgsStartsWith = "WARNING: Type mapping failed;SQLite error";

            if (!ChoWFServerCmdLineArgBuilder.RunAsServer)
            {
                int batchId = _wfExecutionLineArgsObject.BatchId;
                obj.ApplicationBehaviourSettings.SingleInstanceApp = false;
                if (!_wfExecutionLineArgsObject.JobName.IsNullOrWhiteSpace())
                {
                    obj.LogSettings.DoAppendProcessIdToLogDir = false;
                    if (batchId != 0)
                    {
                        obj.LogSettings.LogFolder = !ChoApplication.AppEnvironment.IsNullOrWhiteSpace() ?
                        Path.Combine(ChoPath.AssemblyBaseDirectory, "Logs", ChoApplication.AppEnvironment, batchId.ToString(), _wfExecutionLineArgsObject.JobName) :
                        Path.Combine(ChoPath.AssemblyBaseDirectory, "Logs", batchId.ToString(), _wfExecutionLineArgsObject.JobName);
                    }
                    else
                    {
                        obj.LogSettings.LogFolder = !ChoApplication.AppEnvironment.IsNullOrWhiteSpace() ?
                        Path.Combine(ChoPath.AssemblyBaseDirectory, "Logs", ChoApplication.AppEnvironment, _wfExecutionLineArgsObject.JobName) :
                        Path.Combine(ChoPath.AssemblyBaseDirectory, "Logs", _wfExecutionLineArgsObject.JobName);
                    }

                    obj.LogSettings.LogFileName = "WF_{0}.log".FormatString(_wfExecutionLineArgsObject.JobName);
                }
            }
            else
            {
                //obj.TrayApplicationBehaviourSettings.TurnOn = true;
                //obj.TrayApplicationBehaviourSettings.HideTrayIconWhenMainWindowShown = true;
            }
            base.ApplyGlobalApplicationSettingsOverrides(obj);
        }

        protected override void AfterAppFrxSettingsLoaded(object obj)
        {
            if (obj is ChoAppFrxSettings)
            {
                ((ChoAppFrxSettings)obj).DoNotShowEnvSelectionWnd = true; // !_wfServerCmdLineArgObject.IsController;
                _wfExecutionLineArgsObject = ChoWFServerCmdLineArgBuilder.WFExecutionLineArgsObject;
                _wfServerCmdLineArgsObject = new ChoWFServerCmdLineArgsObject();
            }
            else if (obj is ChoCodeBaseSettings)
            {
                ((ChoCodeBaseSettings)obj).IncludeAppBaseSubDirectories = false;
            }
            base.AfterAppFrxSettingsLoaded(obj);
        }

        #endregion Main Entry

        #region ChoApplicationHost Overrides

        protected override void OnStart(string[] args)
        {
            if (ChoApplication.ApplicationMode == ChoApplicationMode.Console)
            {
                ChoWorkflowServerSettings settings = new ChoWorkflowServerSettings();

                ChoConfigurationObjectEditor wnd = new ChoConfigurationObjectEditor(settings);
                wnd.Title = "Server Settings";
                wnd.ResizeMode = ResizeMode.CanMinimize;

                wnd.ConfigurationObjectEditorCtrl.ShowTitle = false;
                if (!wnd.ShowDialog().Value)
                    return;
            }
            StartAsWorkflowServer();

            base.OnStart(args);

            _idleTask = new ChoIdleTask();
            _idleTask.Start();
        }

        #endregion ChoApplicationHost Overrides

        private void StartAsWorkflowServer()
        {
            MaxNoOfAllowedTasksReached = false;
            _workflowServerSettings = new ChoWorkflowServerSettings();
            ChoWFServerCmdLineArgsObject wfServerCmdLineArgsObject = new ChoWFServerCmdLineArgsObject();

            ChoWSUDPConnectionManager.Instance.StartAsServer();
            _worklowServiceProgramFilePath = Assembly.GetExecutingAssembly().Location;
            ChoTrace.WriteLine("Executable Path: " + _worklowServiceProgramFilePath);
            _concurrencySemaphore = new SemaphoreSlim(_workflowServerSettings.MaxNoOfConcurrentTasks);
            ChoWorkflowDefManager.Initialize();
            ChoWorkflowDefManager.OnWorkflowDefDiscovered += new EventHandler<ChoEventArgs<string[]>>(OnWorkflowDefDiscovered);

            if (wfServerCmdLineArgsObject.IsPeer)
                RunServer(wfServerCmdLineArgsObject.IsPeer);
            else
                RunServer(wfServerCmdLineArgsObject.RunAsPeer);
        }

        private void RunServer(bool isPeer)
        {
            if (!isPeer)
                StartAsController();
            else
                StartAsPeer(_workflowServerSettings.ControllerIPAddress, _workflowServerSettings.Port, _workflowServerSettings.TurnOnUDP);
        }

        private void StartAsPeer(string controllerIPAddress, int port, bool turnOnUDP)
        {
            ResetServer();

            if (_workflowServerSettings.ControllerIPAddress.IsNullOrWhiteSpace())
                throw new ApplicationException("Missing controller IPAddress.");

            ChoWFServerCmdLineArgsObject wfServerCmdLineArgsObject = new ChoWFServerCmdLineArgsObject();
            ChoTrace.WriteLine("Starting as peer.... {0}:{1}".FormatString(controllerIPAddress, port));
            ChoConsole.WriteLine("Starting as peer.... {0}:{1}".FormatString(controllerIPAddress, port));
            ChoWSUDPConnectionManager.Instance.StartAsPeer();

            _dataModel = ChoDataModelFactory.New();

            ForceAbortRunningTasks();

            //Start WFDef Synchronizer
            ChoWFDefsFolderSynchronizer.Value.Start(() => _dataModel.GetWFDefsFileList(), (fileName) => _dataModel.DownloadWFDef(fileName));

            //Start PlugIn Synchronizer
            ChoWFPlugInsFolderSynchronizer.Value.Start(() => _dataModel.GetWFPlugInsFileList(), (fileName) => _dataModel.DownloadWFPlugIn(fileName));

            //Start WFData Synchronizer
            ChoWFDataFolderSynchronizer.Value.Start(() => _dataModel.GetWFDataFileList(), (fileName) => _dataModel.DownloadWFData(fileName));

            _tcpClient = new ChoTcpClient(new IPEndPoint(IPAddress.Parse(controllerIPAddress), port));
            _tcpClient.Connected += (sender, evtArg) =>
            {
                ChoConsole.WriteLine("Connected to controller....");
                ChoTrace.WriteLine("Connected to controller....");

                ChoSubscribeCommand subscribeServerCommand = new ChoSubscribeCommand();
                subscribeServerCommand.HostName = Environment.MachineName;
                subscribeServerCommand.SubscriberType = ChoSubscriberType.Server;
                subscribeServerCommand.ClusterName = wfServerCmdLineArgsObject.ClusterName;
                subscribeServerCommand.IsPrivateNode = wfServerCmdLineArgsObject.IsPrivateNode;

                ChoCommandMessage commandMsg = new ChoCommandMessage();
                commandMsg.Command = subscribeServerCommand;

                ((ChoTcpClient)sender).Send(commandMsg);

                ChoETLFeedbackService.Controller = ((ChoTcpClient)sender);
            };
            _tcpClient.Disconnected += (s, e) =>
            {
                ChoConsole.WriteLine("Disconnected from controller....");
                ChoTrace.WriteLine("Disconnected from controller....");
                //if (turnOnUDP)
                //{
                //    try
                //    {
                //        ((ChoTcpClient)s).Stop();
                //    }
                //    catch { }
                //    Task.Factory.StartNew(() =>
                //    {
                //        ResetServer();
                //        StartAsWorkflowServer();
                //    }).LogExceptions();
                //}
            };

            _tcpClient.OnObjectRecieved = ((st, e) =>
            {
                if (!(e is ChoCommandMessage)) return;
                OnCommandMsgObjectReceived(st, e as ChoCommandMessage);
            });

            _tcpClient.Start();
        }

        private bool _isStarting = false;
        private void StartAsController(bool pingCall = false)
        {
            ResetServer();

            try
            {
                _dataModel = ChoDataModelFactory.New();

                ForceAbortRunningTasks();
                StartDataModelService();

                ChoTrace.WriteLine("Starting as controller.... Port:{0}".FormatString(_workflowServerSettings.Port));
                ChoConsole.WriteLine("Starting as controller.... Port:{0}".FormatString(_workflowServerSettings.Port));
                _tcpServer = new ChoTcpServer(_workflowServerSettings.Port);

                //if (!pingCall)
                //{
                //    _pingAppServer = new ChoPingAppServer(ChoIPAddress.GetLocalIPAddress(), _workflowServerSettings.Port);
                //    _pingAppServer.PingCompleted += (o, e) =>
                //    {
                //        if (!e.IsSuccess)
                //        {
                //            if (!_isStarting)
                //            {
                //                _isStarting = true;
                //                Task.Factory.StartNew(() => StartAsController(true));
                //            }
                //        }
                //    };
                //}

                _wsControllerConnectionManager = new ChoWSControllerConnectionManager(_tcpServer);
                _tcpServer.OnObjectRecieved = ((st, e) =>
                {
                    if (!(e is ChoCommandMessage)) return;
                    OnCommandMsgObjectReceived(st, e as ChoCommandMessage);
                });
                ChoWFPlugInsFolderSynchronizer.Value.SrcChanged += ((o, e) =>
                {
                    _wsControllerConnectionManager.BroadcastCommand(ChoCommandMessage.DefaultWFPlugInsChangedMsg);
                });
                ChoWFDefsFolderSynchronizer.Value.SrcChanged += ((o, e) =>
                {
                    _wsControllerConnectionManager.BroadcastCommand(ChoCommandMessage.DefaultWFDefsChangedMsg);
                });
                ChoWFDataFolderSynchronizer.Value.SrcChanged += ((o, e) =>
                {
                    _wsControllerConnectionManager.BroadcastCommand(ChoCommandMessage.DefaultWFDataChangedMsg);
                });

                _tcpServer.Start();

                ChoWFServerCmdLineArgsObject wfServerCmdLineArgsObject = new ChoWFServerCmdLineArgsObject();
                _wsControllerConnectionManager.AddToMachineHashSet(Environment.MachineName, wfServerCmdLineArgsObject.ClusterName, wfServerCmdLineArgsObject.IsPrivateNode);

                StartMsgBroadcaster(_workflowServerSettings.PollInterval);

                StartClientController();
                StartSchedulerService();

                //if (!pingCall)
                //    _pingAppServer.Start();
            }
            finally
            {
                _isStarting = false;
            }
        }

        private void ResetServer()
        {
            ResetRequested = true;

            try
            {
                _modifiedDate = DateTime.MinValue;

                StopSchedulerService();
                StopMsgBroadcaster();

                try
                {
                    if (_wcfHost != null)
                    {
                        _wcfHost.Close();
                        _wcfHost = null;
                    }
                }
                catch { }

                ChoTaskProcessManager.KillAllTasks();

                if (_wsControllerConnectionManager != null)
                {
                    _wsControllerConnectionManager.Dispose();
                    _wsControllerConnectionManager = null;
                }

                if (_tcpClient != null)
                {
                    _tcpClient.Stop();
                    _tcpClient = null;
                }

                if (_pingAppServer != null)
                {
                    _pingAppServer.Dispose();
                    _pingAppServer = null;
                }

                if (_tcpServer != null)
                {
                    _tcpServer.Stop();
                    _tcpServer = null;
                }
                ChoWFDefsFolderSynchronizer.Value.Stop();
                ChoWFDataFolderSynchronizer.Value.Stop();
                StopClientController();
            }
            finally
            {
                ResetRequested = false;
            }
        }

        private void StopClientController()
        {
            ChoAbortableAsyncResult clientControllerAsyncResult = _clientControllerAsyncResult;
            if (clientControllerAsyncResult == null) return;

            clientControllerAsyncResult.Abort();
        }

        private void StartClientController()
        {
            _clientControllerAsyncResult = ChoActionEx.RunAsync(() =>
            {
                while (true)
                {
                    try
                    {
                        bool isModified = IsMaxNoOfAllowedTasksReached();

                        if (isModified)
                        {
                            if (_wsControllerConnectionManager != null)
                            {
                                _wsControllerConnectionManager.PublishClientCtrlMsgToClients();
                            }
                        }
                    }
                    catch (Exception ex)
                    {
                        ChoTrace.Error(ex);
                    }

                    Thread.Sleep(5000);
                }
            });
        }

        private bool IsMaxNoOfAllowedTasksReached()
        {
            /*
            try
            {
                if (_dataModel != null)
                {
                    ChoTasksDataSet.CHO_TASKSDataTable tasksDataTable = _dataModel.GetAllTasks(null, null, null, ChoDbDateTime.MinValue);
                    if (tasksDataTable.Rows.Count >= _workflowServerSettings.MaxNoOfAllowedTasks)
                    {
                        if (!MaxNoOfAllowedTasksReached)
                        {
                            MaxNoOfAllowedTasksReached = true;
                            return true;
                        }
                    }
                    else
                    {
                        if (MaxNoOfAllowedTasksReached)
                        {
                            MaxNoOfAllowedTasksReached = false;
                            return true;
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                ChoTrace.Error(ex);
            }
            */
            return false;
        }

        private void OnWorkflowDefDiscovered(object sender, ChoEventArgs<string[]> e)
        {
            if (_wsControllerConnectionManager != null)
                _wsControllerConnectionManager.PublishWorkflowListToClients();
        }

        private void RunWorkflow()
        {
            //Debugger.Launch();
            Console.WriteLine("Process Id: {0}".FormatString(Process.GetCurrentProcess().Id));
            if (_wfExecutionLineArgsObject.WFFilePath.IsNullOrWhiteSpace())
            {
                _wfExecutionLineArgsObject.WFFilePath = ChoWorkflowDefManager.GetWorkflowDefFilePath(_wfExecutionLineArgsObject.TaskName);
                //throw new ApplicationException("Missing workflow file path.");
            }

            //ChoTaskGroupCategory taskGroupCategory = new ChoTaskGroupCategory(0, _wfExecutionLineArgsObject.TaskName, _wfExecutionLineArgsObject.ScheduleName);
            //ChoTaskModel.Refresh(taskGroupCategory);

            //if ((ChoETLWorkflowStatus)taskGroupCategory.TaskStatus == ChoETLWorkflowStatus.RUNNING)
            //{
            //    ChoTrace.WriteLine("Task is in running state.");
            //    return;
            //}

            ChoTrace.WriteLine("Running workflow...");
            SetTaskStatus(_wfExecutionLineArgsObject.BatchId, _wfExecutionLineArgsObject.TaskName, _wfExecutionLineArgsObject.ScheduleName, ChoETLWorkflowStatus.RUNNING, _wfExecutionLineArgsObject.UserName, Environment.MachineName, "Preparing the task...");

            StartIPCService();

            using (ChoETLRuntime rt = new ChoETLRuntime())
            {
                rt.EnableFeedbackService = false;

                rt.BeforeWorkflowConstructionStarted += ((sender, e) =>
                {
                    ChoETLWorkflow wf = sender as ChoETLWorkflow;
                    if (wf == null) return;
                    wf.CmdParameters = _wfExecutionLineArgsObject.WFParams;
                });

                rt.AfterWorkflowConstructionComplete += ((sender, e) =>
                {
                    ChoETLActionDef actionDef = new ChoETLActionDef() { Name = "fsStd", ClassTypeName = "ChoWorkflowServer.Model.ChoETLFeedbackService, ChoWorkflowServer.Model", MethodName = "UpdateWorkflowStatus" };
                    rt.AddFeedbackServiceDef(actionDef);
                });

                rt.AfterWorkflowInitialized += ((sender, e) =>
                {
                    ChoETLWorkflow wf = sender as ChoETLWorkflow;
                    if (wf == null) return;

                    //wf.JobName = _taskName;
                    wf.Tag = _wfExecutionLineArgsObject.UserName;
                    wf.BatchId = _wfExecutionLineArgsObject.BatchId;

                    rt.EnableFeedbackService = true;

                    wf.CmdParameters = _wfExecutionLineArgsObject.WFParams;
                });


                _runningWF = rt.Run(_wfExecutionLineArgsObject.WFFilePath, !_wfExecutionLineArgsObject.ResumeTask, ChoJobNameHelper.ToJobName(_wfExecutionLineArgsObject.TaskName, _wfExecutionLineArgsObject.ScheduleName),
                    _wfExecutionLineArgsObject.Timeout <= 0 ? Timeout.Infinite : _wfExecutionLineArgsObject.Timeout, _wfExecutionLineArgsObject.SleepBetweenRetry, _wfExecutionLineArgsObject.MaxNoOfRetry);
                //_runningWF.Start(true);
                _runningWF.Wait();
                _exitCode = _runningWF.ExitCode;
                if (_cts != null)
                    _cts.Cancel();
            }
        }

        internal void PublishServerListToClients()
        {
            if (_wsControllerConnectionManager != null)
                _wsControllerConnectionManager.PublishServerListToClients();
        }

        private void StartIPCService()
        {
            ChoWFServerCmdLineArgsObject cmd = new ChoWFServerCmdLineArgsObject();
            if (cmd.DebugMode)
                return;

            if (_wfExecutionLineArgsObject.TaskName.IsNullOrWhiteSpace()) return;

            _mmr = new ChoMemoryMappedFileReader<ChoCommandMessage>(ChoMemoryMappedFileHelper.GetName("{0}_{1}".FormatString(_wfExecutionLineArgsObject.TaskName, _wfExecutionLineArgsObject.BatchId)));

            CancellationToken token = _cts.Token;
            Task.Factory.StartNew(() =>
                {
                    while (true)
                    {
                        if (token.IsCancellationRequested)
                            break;

                        ChoCommandMessage payload = _mmr.GetData();
                        if (payload != null)
                        {
                            if (payload.Command is ChoTaskActionCommand)
                                ProcessTaskIPCCommand(payload.Command as ChoTaskActionCommand, payload.TimeStamp, payload.User);
                            else
                                ProcessTaskActionCommand(payload);
                        }
                    }
                }, token).LogExceptions();
        }

        private void StopSchedulerService()
        {
            if (_workflowScheduler != null)
            {
                _workflowScheduler.Dispose();
                _workflowScheduler = null;
            }
        }

        private void StartSchedulerService()
        {
            _workflowScheduler = new ChoWorkflowScheduler();
            _dataModel.WorkflowScheduler = _workflowScheduler;

            _workflowScheduler.JobFired += (sender, e) =>
            {
                Tuple<string, string, string> pair = e.Tag as Tuple<string, string, string>;
                if (pair == null)
                {
                    ChoTrace.WriteLine("{0}: Tag is missing".FormatString(e.BatchId));
                    return;
                }
                FireStartTask(e.BatchId, pair.Item1, pair.Item2, pair.Item3, ChoUser.SYSTEM_ACCT, Environment.MachineName);
            };
            _workflowScheduler.JobScheduled += (sender, e) =>
            {
                Tuple<string, string, string> pair = e.Tag as Tuple<string, string, string>;
                if (pair == null) return;

                string taskName = pair.Item1;
                string scheduleName = pair.Item2;
                if (taskName.IsNullOrWhiteSpace() || e.FireTimeUtc == null) return;

                FireScheduledTask(e.BatchId, taskName, scheduleName, e.FireTimeUtc.Value.DateTime);
            };

            _workflowScheduler.JobActivate += (sender, e) =>
            {
                try
                {
                    Tuple<string, string, string> pair = e.Tag as Tuple<string, string, string>;
                    if (pair == null) return;

                    string taskName = pair.Item1;
                    string scheduleName = pair.Item2;
                    if (taskName.IsNullOrWhiteSpace()) return;

                    FireActivateTask(e.BatchId, taskName, scheduleName, e.ForceActivate, ChoUser.SYSTEM_ACCT, Environment.MachineName);
                }
                catch (Exception ex)
                {
                    ChoTrace.Error(ex);
                }
            };

            ChoTasksDataSet.CHO_TASKSDataTable tasksDataTable = _dataModel.GetAllTasks(null, null, null, ChoDbDateTime.MinValue);

            _workflowScheduler.ScheduleWorkflows(tasksDataTable);
            UpdateNextRunTimeForTasks(tasksDataTable);

            //Task.Factory.StartNew(StartWorkflowSchedulerRefresher);
        }

        private void UpdateNextRunTimeForTasks(ChoTasksDataSet.CHO_TASKSDataTable tasksDataTable)
        {
            foreach (ChoTasksDataSet.CHO_TASKSRow row in tasksDataTable)
            {
                try
                {
                    SetTaskStatus(row.BATCH_ID, row.TASK_NAME, row.SCHEDULE_NAME, (ChoETLWorkflowStatus)row.TASK_STATUS, ChoUser.SYSTEM_ACCT, Environment.MachineName, null, row.NEXT_RUN_TIME);
                }
                catch (Exception ex)
                {
                    ChoTrace.Error(ex);
                }
            }
        }

        private ChoTasksDataSet.CHO_TASKSDataTable ForceAbortRunningTasks()
        {
            ChoTasksDataSet.CHO_TASKSDataTable tasksDataTable = _dataModel.GetAllTasks(null, null, _machineName, ChoDbDateTime.MinValue);

            //Any running tasks, reset them to ABORTED
            if (tasksDataTable.Count > 0)
            {
                foreach (ChoTasksDataSet.CHO_TASKSRow row in tasksDataTable)
                {
                    if (!row.IsTASK_STATUSNull() && (ChoETLWorkflowStatus)row.TASK_STATUS == ChoETLWorkflowStatus.RUNNING)
                        SetTaskStatus(row.BATCH_ID, row.TASK_NAME, row.SCHEDULE_NAME, ChoETLWorkflowStatus.ABORTED, ChoUser.SYSTEM_ACCT, Environment.MachineName,
                            "Workflow abandoned with RUNNING state. Resetted to ABORTED state by system.");
                }
            }

            return tasksDataTable;
        }

        private void StartWorkflowSchedulerRefresher()
        {
            DateTime currentDateTime = ChoDbDateTime.MinValue;
            ChoTrace.WriteLine("Starting workflow scheduler refresher...");

            try
            {
                while (true)
                {
                    Thread.Sleep(30 * 1000);

                    ChoTasksDataSet.CHO_TASKSDataTable tasksDataTable = new ChoTasksDataSet.CHO_TASKSDataTable();

                    foreach (ChoTasksDataSet.CHO_TASKSRow row in _dataModel.GetAllTasks(null, null, null, currentDateTime))
                    {
                        try
                        {
                            _workflowScheduler.RemoveJobFromScheduler(row.TASK_ID, row.TASK_NAME, row.SCHEDULE_NAME);
                            ChoTrace.WriteLine("Removing '{0}' from scheduler via refresher...".FormatString(ChoJobNameHelper.ToJobName(row.TASK_NAME, row.SCHEDULE_NAME)));
                        }
                        catch { }
                    }

                    currentDateTime = DateTime.Now;
                }
            }
            catch { }
        }

        private void StartDataModelService()
        {
            Uri baseAddress = new Uri(_workflowServerSettings.DataModelEndPointUrl);
            _wcfHost = new ServiceHost(ChoDataModelFactory.DataModelType, baseAddress);

            //Enable metadata exchange
            ServiceMetadataBehavior smb = new ServiceMetadataBehavior();
            smb.HttpGetEnabled = true;
            _wcfHost.Description.Behaviors.Add(smb);

            ServiceThrottlingBehavior throttleBehaviour = new ServiceThrottlingBehavior();
            throttleBehaviour.MaxConcurrentSessions = 100;
            throttleBehaviour.MaxConcurrentCalls = 25;
            throttleBehaviour.MaxConcurrentInstances = 25;
            _wcfHost.Description.Behaviors.Add(throttleBehaviour);

            _wcfHost.Description.Behaviors.Remove(typeof(ServiceDebugBehavior));
            _wcfHost.Description.Behaviors.Add(new ServiceDebugBehavior() { IncludeExceptionDetailInFaults = true });

            _wcfHost.Open();
            ChoTrace.WriteLine("Data Model Service hosted at {0}...".FormatString(baseAddress));
        }

        internal bool FireScheduledTask(int batchId, string taskName, string scheduleName, DateTime nextRunTime)
        {
            try
            {
                if (taskName.IsNullOrWhiteSpace()) return false;
                ChoTaskGroupCategory cat = GetTask(taskName, scheduleName);
                ChoETLWorkflowStatus currentStatus = cat.WorkflowStatus;
                if (batchId == 0)
                    batchId = cat.BatchId;
                SetTaskStatus(batchId, taskName, scheduleName, currentStatus, ChoUser.SYSTEM_ACCT, Environment.MachineName, String.Empty, nextRunTime.ToString("G"));
            }
            catch (Exception ex)
            {
                ChoTrace.Error(ex);
            }

            return false;
        }

        internal bool FireActivateTask(int batchId, string taskName, string scheduleName, bool forceActivate, string user, string originHostName)
        {
            try
            {
                if (taskName.IsNullOrWhiteSpace()) return false;
                ChoTaskGroupCategory cat = GetTask(taskName, scheduleName);
                ChoETLWorkflowStatus currentStatus = cat.WorkflowStatus;
                if (batchId == 0)
                    batchId = cat.BatchId;

                //if (cat.MachineName != _machineName)
                //    return false;

                cat.MachineName = ChoClusterManager.Default.Resolve(cat.MachineName);

                if (cat.BatchId != batchId)
                {
                    if (currentStatus == ChoETLWorkflowStatus.SUCCESS
                        || currentStatus == ChoETLWorkflowStatus.FAILED
                        || currentStatus == ChoETLWorkflowStatus.ABORTED
                        || currentStatus == ChoETLWorkflowStatus.TIMEDOUT
                        || currentStatus == ChoETLWorkflowStatus.STOPPED
                        )
                    {
                        //SetTaskStatus(batchId, taskName, scheduleName, ChoETLWorkflowStatus.ACTIVE, ChoUser.SYSTEM_ACCT, Environment.MachineName, String.Empty);
                        SetTaskStatus(batchId, taskName, scheduleName, ChoETLWorkflowStatus.ACTIVE, user, originHostName, String.Empty, null, cat.MachineName);
                        FireActivateDependentTasks(batchId, taskName, scheduleName, user, originHostName);
                        return true;
                    }
                    else if (/*forceActivate
                        || */ currentStatus == ChoETLWorkflowStatus.ACTIVE)
                    {
                        //SetTaskStatus(batchId, taskName, scheduleName, ChoETLWorkflowStatus.ACTIVE, ChoUser.SYSTEM_ACCT, Environment.MachineName, String.Empty);
                        SetTaskStatus(batchId, taskName, scheduleName, ChoETLWorkflowStatus.ACTIVE, user, originHostName, String.Empty, null, cat.MachineName);
                        FireActivateDependentTasks(batchId, taskName, scheduleName, user, originHostName);
                        return true;
                    }
                }
            }
            catch (Exception ex)
            {
                ChoTrace.Error(ex);
            }

            return false;
        }

        private void FireActivateDependentTasks(int batchId, string taskName, string scheduleName, string user, string originHostName)
        {
            _workflowScheduler.FireActivateDependentTasks(batchId, taskName, scheduleName, user, originHostName);
        }

        internal void FireStartTask(int batchId, string taskName, string scheduleName, string wfName, string user, string originHostName)
        {
            try
            {
                ChoCommandMessage cmdMsg = ChoCommandMessage.New();
                cmdMsg.User = user;
                cmdMsg.HostName = cmdMsg.OriginHostName = originHostName;

                ChoTaskGroupCategory taskGroupCategory = new ChoTaskGroupCategory(0, taskName, scheduleName, null);
                ChoTaskModel.Refresh(taskGroupCategory);

                cmdMsg.Command = new ChoTaskActionCommand()
                {
                    CommandType = ChoTaskActionType.StartTask,
                    TaskName = taskName,
                    ScheduleName = scheduleName,
                    WFName = wfName,
                    BatchId = batchId,
                    RunDependentTasks = true,
                    Timeout = taskGroupCategory.Timeout == null ? 0 : taskGroupCategory.Timeout.Value,
                    SleepBetweenRetry = taskGroupCategory.SleepBetweenRetry == null ? 0 : taskGroupCategory.SleepBetweenRetry.Value,
                    MaxNoOfRetry = taskGroupCategory.MaxNoOfRetry == null ? 0 : taskGroupCategory.MaxNoOfRetry.Value,
                    CommandParameters = taskGroupCategory.CmdParameters
                };
                ProcessTaskActionCommand(cmdMsg);
            }
            catch (Exception ex)
            {
                ChoTrace.Error(ex);
            }
        }

        private void StopMsgBroadcaster()
        {
            ChoDataModelEvents.Reset();

            if (_msgBroadcastAsyncResult != null)
            {
                _msgBroadcastAsyncResult.Abort();
                _msgBroadcastAsyncResult = null;
            }
        }

        [ChoStreamProfile("MsgBroadCaster")]
        private void StartMsgBroadcaster(int pollInterval)
        {
            ChoDataModelEvents.OnTaskChanged += ((target, args) =>
                {
                    ChoTcpServer tcpServer = _tcpServer;
                    if (tcpServer == null) return;

                    try
                    {
                        ChoCommandMessage commandRequest = args.Value;
                        tcpServer.Broadcast(commandRequest);
                    }
                    catch (Exception ex)
                    {
                        ChoProfile.Default.Append(ex);
                    }
                });
            ChoDataModelEvents.OnCalendarChanged += ((target, args) =>
            {
                ChoTcpServer tcpServer = _tcpServer;
                if (tcpServer == null) return;

                try
                {
                    ChoCommandMessage commandRequest = args.Value;
                    tcpServer.Broadcast(commandRequest);
                }
                catch (Exception ex)
                {
                    ChoProfile.Default.Append(ex);
                }
            });
            ChoDataModelEvents.OnUserACLChanged += ((target, args) =>
            {
                ChoTcpServer tcpServer = _tcpServer;
                if (tcpServer == null) return;

                try
                {
                    ChoCommandMessage commandRequest = args.Value;
                    tcpServer.Broadcast(commandRequest);
                }
                catch (Exception ex)
                {
                    ChoProfile.Default.Append(ex);
                }
            });

            _msgBroadcastAsyncResult = ChoActionEx.RunAsync(() =>
            {
                ChoTasksDataSet.CHO_TASKSDataTable tasksDataTable = new ChoTasksDataSet.CHO_TASKSDataTable();
                string debugMsg = null;

                using (ChoStreamProfile profile = new ChoStreamProfile("wfStatusMsgs", "Following messages are sent to clients...", ChoProfile.NULL))
                {
                    while (true)
                    {
                        tasksDataTable.Clear();
                        tasksDataTable = _dataModel.GetAllTasks(null, null, null /*_machineName */, _modifiedDate);

                        if (tasksDataTable.Count > 0)
                        {
                            foreach (ChoTasksDataSet.CHO_TASKSRow row in tasksDataTable)
                            {
                                if (row == null && row.IsMODIFIED_DATENull())
                                    continue;

                                if (row.MODIFIED_DATE > _modifiedDate)
                                    _modifiedDate = row.MODIFIED_DATE;

                                try
                                {
                                    ChoCommandMessage payload = ChoCommandMessage.New();
                                    payload.Universalize();

                                    ChoTaskMsgPayload msg = ChoTaskMsgPayload.New(row);

                                    //if (msg.ModifiedDate > _modifiedDate)
                                    //    _modifiedDate = msg.ModifiedDate;

                                    debugMsg = ChoObject.ToString(msg);
                                    profile.Info(debugMsg);

                                    payload.Command = msg;
                                    _wsControllerConnectionManager.SendCommandToAllClients(payload);
                                }
                                catch (Exception ex)
                                {
                                    ChoProfile.Default.Append(ex);
                                }
                            }
                            _modifiedDate = _modifiedDate.AddMilliseconds(1);
                        }

                        Thread.Sleep(pollInterval);
                    }
                }
            });
        }
        protected override void OnStop()
        {
            ResetServer();

            base.OnStop();

            if (_idleTask != null)
            {
                _idleTask.Stop();
                _idleTask = null;
            }
        }

        internal void SetTaskStatus(int batchId, string taskName, string scheduleName, Cinchoo.Core.ETL.ChoETLWorkflowStatus taskStatus, 
            string modifiedBy, string modifiedByHostName, string logMsg = null, string nextRunTime = null, string machineName = null)
        {
            if (_dataModel == null) return;

            _dataModel.SetTaskStatus(batchId, taskName, scheduleName, taskStatus, modifiedBy, modifiedByHostName, logMsg, nextRunTime, machineName);
            //rt.QueueService.Enqueue(() => _dataModel.SetTaskStatus(batchId, taskName, taskStatus, modifiedBy, logMsg)).AsyncWaitHandle.WaitOne();
        }

        internal string GetTaskCmdParameters(string taskName, string scheduleName)
        {
            ChoTaskGroupCategory taskGroupCategory = new ChoTaskGroupCategory(0, taskName, scheduleName, null);
            ChoTaskModel.Refresh(taskGroupCategory);

            return taskGroupCategory.CmdParameters;
        }

        internal string GetTaskBatchCmdParameters(string taskName, string scheduleName)
        {
            ChoTaskGroupCategory taskGroupCategory = new ChoTaskGroupCategory(0, taskName, scheduleName, null);
            ChoTaskModel.Refresh(taskGroupCategory);

            return taskGroupCategory.BatchCmdParameters;
        }

        internal Tuple<string, string> GetTaskMachineName(string taskName, string scheduleName)
        {
            ChoTaskGroupCategory taskGroupCategory = new ChoTaskGroupCategory(0, taskName, scheduleName, null);
            ChoTaskModel.Refresh(taskGroupCategory);

            return new Tuple<string, string>(taskGroupCategory.MachineName, taskGroupCategory.BatchMachineName);
        }

        internal int GetTaskBatchId(string taskName, string scheduleName)
        {
            ChoTaskGroupCategory taskGroupCategory = new ChoTaskGroupCategory(0, taskName, scheduleName, null);
            ChoTaskModel.Refresh(taskGroupCategory);
            return taskGroupCategory.BatchId;
        }

        internal ChoTaskGroupCategory GetTask(string taskName, string scheduleName)
        {
            ChoTaskGroupCategory taskGroupCategory = new ChoTaskGroupCategory(0, taskName, scheduleName, null);
            ChoTaskModel.Refresh(taskGroupCategory);
            return taskGroupCategory;
        }

        internal ChoETLWorkflowStatus GetTaskStatus(string taskName, string scheduleName)
        {
            //IChoAsyncResult r = rt.QueueService.Enqueue<ChoETLWorkflowStatus>(() =>
            //    {
            ChoTaskGroupCategory TaskGroupCategory = new ChoTaskGroupCategory(0, taskName, scheduleName, null);
                    ChoTaskModel.Refresh(TaskGroupCategory);

                    return (ChoETLWorkflowStatus)TaskGroupCategory.TaskStatus;
            //    });

            //r.AsyncWaitHandle.WaitOne();
            //return (ChoETLWorkflowStatus)r.Result;
        }

        internal int GetTaskExitCode(string taskName, string scheduleName)
        {
            ChoTaskGroupCategory TaskGroupCategory = new ChoTaskGroupCategory(0, taskName, scheduleName, null);
            ChoTaskModel.Refresh(TaskGroupCategory);

            return TaskGroupCategory.ExitCode;
        }
    }
}
